-module(rtps_writer_reader_SUITE).

%%% ============================================================================
%%% This test suite uses all legal combinations of writers and readers
%%% and checks for proper data transfer .
%%% ============================================================================

-include_lib("common_test/include/ct.hrl").
-include("../src/rtps.hrl").

-export([all/0, suite/0, groups/0,
	 init_per_suite/1, end_per_suite/1, 
	 init_per_group/2, end_per_group/2,
	 init_per_testcase/2, end_per_testcase/2]).

-compile(export_all).

%% To run the tests using a remote system, install RTPS on the remote
%% system and configure ssh in such a way that the user executing this
%% test can login without password authentication, i.e. use preshared
%% keys (man ssh-copy-id).
%% If epmd isn't running yet on the remote host, you get some
%% errors and it will be started auto-magically. If not, start
%% epmd by hand.

%% -define(REMOTE_HOST, undefined). % when there is no remote system for running the tests
-define(REMOTE_HOST, rpi3). % A Raspberry Pi 3 Model B+
%% -define(REMOTE_HOST, laptop). % A Dell laptop doing much better
-define(REMOTE_NODE_NAME, b).

-define(DOMAIN_ID, 0).
-define(WRITER_GUID_PREFIX, <<1:96>>).
-define(WRITER_ENTITY_KEY, <<1:24>>).
-define(WRITER_GUID, #guid{guid_prefix = ?WRITER_GUID_PREFIX,
			   entity_id = #entity_id{type = user,
						  kind = writer_no_key,
						  key = ?WRITER_ENTITY_KEY}}).
-define(READER_GUID_PREFIX, <<2:96>>).
-define(READER_ENTITY_KEY, <<2:24>>).
-define(READER_GUID, #guid{guid_prefix = ?READER_GUID_PREFIX,
			   entity_id = #entity_id{type = user,
						  kind = reader_no_key,
						  key = ?READER_ENTITY_KEY}}).

-define(N, 100). % Number of cache changes to transfer
-define(DATA_VALUE, <<"This is a test!">>).

suite() ->
    [{timetrap, 1000}].

all() -> 
    [{group, local},
     {group, remote}
    ].

groups() ->
    [{local,           [], [{group, writers_readers}]},
     {remote,          [], [{group, writers_readers}]},
     {writers_readers, [], [wslb_rslb, 
			    wdlb_rsfb,
			    wslrpmt_rslb,
			    wslrpmt_rsfb,
			    wslrpmf_rsfr,
			    wslrpmt_rsfr,
			    wsfb_rslb,
			    wsfb_rsfb,
			    wsfrpmt_rslb,
			    wsfrpmt_rsfb,
			    wsfrpmf_rsfr,
			    wsfrpmt_rsfr
			   ]}
    ].

init_per_suite(Config) ->
    Config.

end_per_suite(_Config) ->
    ok.

%%%===================================================================
%%% Init per group
%%%===================================================================

init_per_group(remote) ->
    [{timetrap, 10000}];
init_per_group(_Group) ->
    [].

init_per_group(local, Config) ->
    %% Running the tests locally, requires the UDP socket to operate
    %% in multicast loopback mode.
    ok = application:start(rtps),
    Opts = [{multicast_loop, true}],
    {ok, _} = add_domain(local, ?DOMAIN_ID, Opts),
    [{nodes, {local, local}} | Config];
init_per_group(remote, _Config) when ?REMOTE_HOST =:= undefined ->
    {skip, "Remote not configured"};
init_per_group(remote, Config) ->
    Writer_node = local,
    Reader_node = start_node(?REMOTE_HOST, ?REMOTE_NODE_NAME),
    ok = start_application(Writer_node),
    ok = start_application(Reader_node),
    Opts = [],
    {ok, _} = add_domain(Writer_node, ?DOMAIN_ID, Opts),
    {ok, _} = add_domain(Reader_node, ?DOMAIN_ID, Opts),
    [{nodes, {Writer_node, Reader_node}}, {opts, []} | Config];
init_per_group(_Group, Config) ->
    Config.

%%%===================================================================
%%% End per group
%%%===================================================================

end_per_group(remote) ->
    [{timetrap, 10000}];
end_per_group(_Group) ->
    [].

end_per_group(local, _Config) ->
    ok = del_domain(local, ?DOMAIN_ID),
    ok = stop_application(local);
end_per_group(remote, Config) ->
    {Writer_node, Reader_node} = ?config(nodes, Config),
    ok = del_domain(Writer_node, ?DOMAIN_ID),
    ok = del_domain(Reader_node, ?DOMAIN_ID),
    ok = stop_application(Writer_node),
    ok = stop_application(Reader_node),
    ok = stop_node(Writer_node),
    ok = stop_node(Reader_node);
end_per_group(_Group, _Config) ->
    ok.

%%%===================================================================
%%% Init per case
%%%===================================================================

init_per_testcase(_Case, Config) ->
    {Writer_node, Reader_node} = ?config(nodes, Config),
    {ok, _} = add_participant(Writer_node, ?DOMAIN_ID, ?WRITER_GUID_PREFIX),
    {ok, _} = add_participant(Reader_node, ?DOMAIN_ID, ?READER_GUID_PREFIX),
    Config.

%%%===================================================================
%%% End per case
%%%===================================================================

end_per_testcase(_Case, Config) ->
    {Writer_node, Reader_node} = ?config(nodes, Config),
    ok = del_participant(Writer_node, ?DOMAIN_ID, ?WRITER_GUID_PREFIX),
    ok = del_participant(Reader_node, ?DOMAIN_ID, ?READER_GUID_PREFIX),
    ok.

%%%===================================================================
%%% Test cases
%%%===================================================================

wslb_rslb(Config) -> 
    ct:comment("writer: stateless, best_effort - reader: stateless, best_effort"),
    W_opts = [{state_type, stateless}, {reliability_level, best_effort}],
    R_opts = [{state_type, stateless}, {reliability_level, best_effort}],
    check_writer_reader(Config, W_opts, R_opts).

wdlb_rsfb(Config) ->
    ct:comment("writer: stateless, best_effort - reader: stateful, best_effort"),
    W_opts = [{state_type, stateless}, {reliability_level, best_effort}],
    R_opts = [{state_type, stateful}, {reliability_level, best_effort}],
    check_writer_reader(Config, W_opts, R_opts).

wslrpmt_rslb(Config) ->
    ct:comment("writer: stateless, reliable, push_mode, true - reader: stateless, best_effort"),
    W_opts = [{state_type, stateless}, {reliability_level, reliable}, {push_mode, true}],
    R_opts = [{state_type, stateless}, {reliability_level, best_effort}],
    check_writer_reader(Config, W_opts, R_opts).

wslrpmt_rsfb(Config) ->
    ct:comment("writer: stateless, reliable, push_mode, true - reader: stateful, best_effort"),
    W_opts = [{state_type, stateless}, {reliability_level, reliable}, {push_mode, true}],
    R_opts = [{state_type, stateful}, {reliability_level, best_effort}],
    check_writer_reader(Config, W_opts, R_opts).

wslrpmf_rsfr(Config) ->
    ct:comment("writer: stateless, reliable, push_mode, false - reader: stateful, reliable"),
    W_opts = [{state_type, stateless}, {reliability_level, reliable}, {push_mode, false}],
    R_opts = [{state_type, stateful}, {reliability_level, reliable}],
    check_writer_reader(Config, W_opts, R_opts).

wslrpmt_rsfr(Config) ->
    ct:comment("writer: stateless, reliable, push_mode, true - reader: stateful, reliable"),
    W_opts = [{state_type, stateless}, {reliability_level, reliable}, {push_mode, true}],
    R_opts = [{state_type, stateful}, {reliability_level, reliable}],
    check_writer_reader(Config, W_opts, R_opts).

wsfb_rslb(Config) ->
    ct:comment("writer: stateful, best_effort - reader: stateless, best_effort"),
    W_opts = [{state_type, stateful}, {reliability_level, best_effort}],
    R_opts = [{state_type, stateless}, {reliability_level, best_effort}],
    check_writer_reader(Config, W_opts, R_opts).

wsfb_rsfb(Config) ->
    ct:comment("writer: stateful, best_effort - reader: stateful, best_effort"),
    W_opts = [{state_type, stateful}, {reliability_level, best_effort}],
    R_opts = [{state_type, stateful}, {reliability_level, best_effort}],
    check_writer_reader(Config, W_opts, R_opts).

wsfrpmt_rslb(Config) ->
    ct:comment("writer: stateful, reliable, push_mode, true - reader: stateless, best_effort"),
    W_opts = [{state_type, stateful}, {reliability_level, reliable}, {push_mode, true}],
    R_opts = [{state_type, stateless}, {reliability_level, best_effort}],
    check_writer_reader(Config, W_opts, R_opts).

wsfrpmt_rsfb(Config) ->
    ct:comment("writer: stateful, reliable, push_mode, true - reader: stateful, best_effort"),
    W_opts = [{state_type, stateful}, {reliability_level, reliable}, {push_mode, true}],
    R_opts = [{state_type, stateful}, {reliability_level, best_effort}],
    check_writer_reader(Config, W_opts, R_opts).

wsfrpmf_rsfr(Config) ->
    ct:comment("writer: stateful, reliable, push_mode, false - reader: stateful, reliable"),
    W_opts = [{state_type, stateful}, {reliability_level, reliable}, {push_mode, false}],
    R_opts = [{state_type, stateful}, {reliability_level, reliable}],
    check_writer_reader(Config, W_opts, R_opts).

wsfrpmt_rsfr(Config) ->
    ct:comment("writer: stateful, reliable, push_mode, true - reader: stateful, reliable"),
    W_opts = [{state_type, stateful}, {reliability_level, reliable}, {push_mode, true}],
    R_opts = [{state_type, stateful}, {reliability_level, reliable}],
    check_writer_reader(Config, W_opts, R_opts).
							 
%%%===================================================================
%%% Internal functions
%%%===================================================================

check_writer_reader(Config, W_opts, R_opts) ->
    {Writer_node, Reader_node} = ?config(nodes, Config),
    Reader_pid = start_reader(Reader_node, R_opts),
    Writer_pid = start_writer(Writer_node, W_opts),
    Cache_changes = cache_changes(Writer_node, Writer_pid, ?N),
    ok = add_change(Writer_node, Writer_pid, Cache_changes),
    available_changes(Reader_node, Reader_pid, ?N),
    del_writer(Writer_node, ?DOMAIN_ID, ?WRITER_GUID_PREFIX, ?WRITER_ENTITY_KEY),
    del_reader(Reader_node, ?DOMAIN_ID, ?READER_GUID_PREFIX, ?READER_ENTITY_KEY).    

start_writer(Writer_node, W_opts) ->
    Writer_state_type = proplists:get_value(state_type, W_opts),
    {ok, _} = add_writer(Writer_node, ?DOMAIN_ID, ?WRITER_GUID_PREFIX, ?WRITER_ENTITY_KEY, W_opts),
    Writer_pid = whereis_writer(Writer_node, ?DOMAIN_ID, ?WRITER_GUID_PREFIX, ?WRITER_ENTITY_KEY),
    %% NB: A reminder how to set the QoS for a writer. The QoS values
    %% will be transported to the reader with every DATA message sent.
    rtps_writer:set_qos(Writer_pid, []),
    case Writer_state_type of
	stateful ->
	    {ok, _} = matched_reader_add(Writer_node, Writer_pid, ?READER_GUID);
	_ ->
	    ok
    end,
    Writer_pid.

cache_changes(Writer_node, Writer_pid, N) ->
    [begin
	 {ok, Change} = new_change(Writer_node, Writer_pid),
	 Change
     end
     || _I <- lists:seq(1, N)].
    
new_change(local, Writer_pid) ->
    rtps_writer:new_change(Writer_pid, kind, ?DATA_VALUE, undefined);
new_change(Node, Writer_pid) ->
    ct_rpc:call(Node, rtps_writer, new_change, [Writer_pid, kind, ?DATA_VALUE, undefined]).

add_change(local, Writer_pid, Cache_changes) ->
    rtps_writer:add_change(Writer_pid, Cache_changes);
add_change(Node, Writer_pid, Cache_changes) ->
    ct_rpc:call(Node, rtps_writer, add_change, [Writer_pid, Cache_changes]).

available_changes(Node, Reader_pid, N) ->
    case available_changes(Node, Reader_pid) of
	Cache_changes when length(Cache_changes) =:= N ->
	    Cache_changes;
	_ ->
	    timer:sleep(10),
	    available_changes(Node, Reader_pid, N)
    end.

available_changes(local, Reader_pid) ->
    rtps_reader:available_changes(Reader_pid);
available_changes(Node, Reader_pid) ->
    ct_rpc:call(Node, rtps_reader, available_changes, [Reader_pid]).

start_reader(Reader_node, R_opts) ->
    Reader_state_type = proplists:get_value(state_type, R_opts),
    {ok, _} = add_reader(Reader_node, ?DOMAIN_ID, ?READER_GUID_PREFIX, ?READER_ENTITY_KEY, R_opts),
    Reader_pid = whereis_reader(Reader_node, ?DOMAIN_ID, ?READER_GUID_PREFIX, ?READER_ENTITY_KEY),
    case Reader_state_type of
	stateful ->
	    {ok, _} = matched_writer_add(Reader_node, Reader_pid, ?WRITER_GUID);
	_ ->
	    ok
    end,
    Reader_pid.

add_domain(Node, Domain_id) ->
    add_domain(Node, Domain_id, []).

add_domain(local, Domain_id, Opts) ->
    rtps:add_domain(Domain_id, Opts);
add_domain(Node, Domain_id, Opts) ->
    ct_rpc:call(Node, rtps, add_domain, [Domain_id, Opts]).

del_domain(local, Domain_id) ->
    rtps:del_domain(Domain_id);
del_domain(Node, Domain_id) ->
    ct_rpc:call(Node, rtps, del_domain, [Domain_id]).

add_participant(local, Domain_id, Guid_prefix) ->
    rtps:add_participant(Domain_id, Guid_prefix);
add_participant(Node, Domain_id, Guid_prefix) ->
    ct_rpc:call(Node, rtps, add_participant, [Domain_id, Guid_prefix]).
    
del_participant(local, Domain_id, Guid_prefix) ->
    rtps:del_participant(Domain_id, Guid_prefix);
del_participant(Node, Domain_id, Guid_prefix) ->
    ct_rpc:call(Node, rtps, del_participant, [Domain_id, Guid_prefix]).

add_writer(local, Domain_id, Guid_prefix, Entity_id, Opts) ->
    rtps:add_writer(Domain_id, Guid_prefix, Entity_id, Opts);
add_writer(Node, Domain_id, Guid_prefix, Entity_id, Opts) ->
    ct_rpc:call(Node, rtps, add_writer, [Domain_id, Guid_prefix, Entity_id, Opts]).

del_writer(local, Domain_id, Guid_prefix, Entity_id) ->
    rtps:del_writer(Domain_id, Guid_prefix, Entity_id);
del_writer(Node, Domain_id, Guid_prefix, Entity_id) ->
    ct_rpc:call(Node, rtps, del_writer, [Domain_id, Guid_prefix, Entity_id]).

add_reader(local, Domain_id, Guid_prefix, Entity_id, Opts) ->
    rtps:add_reader(Domain_id, Guid_prefix, Entity_id, Opts);
add_reader(Node, Domain_id, Guid_prefix, Entity_id, Opts) ->
    ct_rpc:call(Node, rtps, add_reader, [Domain_id, Guid_prefix, Entity_id, Opts]).

del_reader(local, Domain_id, Guid_prefix, Entity_id) ->
    rtps:del_reader(Domain_id, Guid_prefix, Entity_id);
del_reader(Node, Domain_id, Guid_prefix, Entity_id) ->
    ct_rpc:call(Node, rtps, del_reader, [Domain_id, Guid_prefix, Entity_id]).

whereis_writer(local, Domain_id, Guid_prefix, Entity_id) ->
    rtps:whereis_writer(Domain_id, Guid_prefix, Entity_id);
whereis_writer(Node, Domain_id, Guid_prefix, Entity_id) ->
    ct_rpc:call(Node, rtps, whereis_writer, [Domain_id, Guid_prefix, Entity_id]).

whereis_reader(local, Domain_id, Guid_prefix, Entity_id) ->
    rtps:whereis_reader(Domain_id, Guid_prefix, Entity_id);
whereis_reader(Node, Domain_id, Guid_prefix, Entity_id) ->
    ct_rpc:call(Node, rtps, whereis_reader, [Domain_id, Guid_prefix, Entity_id]).

matched_writer_add(local, Reader_pid, ?WRITER_GUID) ->
    rtps_reader:matched_writer_add(Reader_pid, ?WRITER_GUID);
matched_writer_add(Node, Reader_pid, ?WRITER_GUID) ->
    {ok, _} = ct_rpc:call(Node, rtps_reader, matched_writer_add, [Reader_pid, ?WRITER_GUID]).

matched_reader_add(local, Writer_pid, ?READER_GUID) ->
    rtps_writer:matched_reader_add(Writer_pid, ?READER_GUID);
matched_reader_add(Node, Writer_pid, ?READER_GUID) ->
    {ok, _} = ct_rpc:call(Node, rtps_writer, matched_reader_add, [Writer_pid, ?READER_GUID]).


info(local, Domain_id, Guid_prefix) ->
    rtps:info(Domain_id, Guid_prefix);
info(Node, Domain_id, Guid_prefix) ->
    ct_rpc:call(Node, rtps, info, [Domain_id, Guid_prefix]).
    
remotes(Node, Domain_id, Guid_prefix) ->
    Info = info(Node, Domain_id, Guid_prefix),
    proplists:get_value(remotes, Info).

check_conn(Node, Domain_id, Guid_prefix1, Guid_prefix2) ->
    Remotes = remotes(Node, Domain_id, Guid_prefix1),
    case lists:member(Guid_prefix2, Remotes) of
	true ->
	    ok;
	false ->
	    timer:sleep(100),
	    check_conn(Node, Domain_id, Guid_prefix1, Guid_prefix2)
    end.

start_application(local) ->
    application:start(rtps);
start_application(Node) ->
    ct_rpc:call(Node, application, start, [rtps]).

stop_application(local) ->
    application:stop(rtps);
stop_application(Node) ->
    ct_rpc:call(Node, application, stop, [rtps]).

start_node(Host, Node_name) ->
    case net_kernel:getopts(node(), []) of
	ignored ->
	    {ok, _} = net_kernel:start([?MODULE, shortnames]);
	_ ->
	    ok
    end,
    Here = filename:dirname(code:which(?MODULE)),
    Ebin = filename:join([Here, "..", "ebin"]),
    Paths = [Here, Ebin],
    {ok, Node} = ct_slave:start(Host, Node_name, [{monitor_master, true}]),
    ok = ct_rpc:call(Node, code, add_pathsa, [Paths]),
    Node.

stop_node(local) ->
    ok;
stop_node(Node) ->
    {ok, Node} = ct_slave:stop(Node),
    ok.
